package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.mapper.ApArticleConfigMapper;
import com.heima.common.constants.message.WmNewsMessageConstants;
import com.heima.model.article.pojo.ApArticleConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;

/**
 * @author Ronaldo
 * @date 2021/8/10
 */
@Component
public class ArticleDownOrUpListener {

    @Autowired
    private ApArticleConfigMapper apArticleConfigMapper;

    @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
    public void articleDownOrUp(ConsumerRecord<String,String> consumerRecord){
        Optional<ConsumerRecord<String, String>> optional = Optional.ofNullable(consumerRecord);
        optional.ifPresent(record ->{
            String value = record.value();
            Map map = JSON.parseObject(value, Map.class);
            Long articleId = Long.valueOf(map.get("articleId").toString());
            Short enable = Short.valueOf(map.get("enable").toString());
            boolean isDown = false;
            if (enable == 0){
                isDown = true;
            }

            ApArticleConfig apArticleConfig = new ApArticleConfig();
            apArticleConfig.setIsDown(isDown);
            apArticleConfigMapper.update(apArticleConfig,Wrappers.<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId,articleId));
        });
    }
}
